package gupao.concurrency.api.sample;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * API并发调用线程池服务,当线程池没有空闲线程运行客户端提交的任务时,有2种运行策略在服务初始化时候可以指定：
 * <ul>
 * <li>
 * {@link ApiExecutorPolicy#CALLER_RUN_POLICY} 客户端线程自己运行任务
 * <li>
 * {@link ApiExecutorPolicy#REJECT_EXCEPTION_POLICY} 抛出
 * {@code RejectedExecutionException},这种模式下，客户端需要自己处理异常
 * </ul>
 * 
 * @author liwu
 */
public class ApiExecutorService
{

	public static class ApiFutureTask<T> extends FutureTask<T> implements ApiFuture<T>
	{

		private static final Method awaitDoneHack;

		static {
			try {
				awaitDoneHack = FutureTask.class.getDeclaredMethod("awaitDone", boolean.class, long.class);
				awaitDoneHack.setAccessible(true);
			}
			catch (NoSuchMethodException | SecurityException e) {
				throw new BugError("can't find method awaitDone(Boolean,Long) in " + FutureTask.class.getName(), e);
			}
		}

		private List<ApiInvokeInfo> apiInvokeInfo = new ArrayList<>();

		public ApiFutureTask(Callable<T> callable)
		{
			super(callable);
		}

		/**
		 * Waits if necessary for the computation to complete, and then retrieves its result.
		 * 
		 * @see FutureTask#get()
		 */
		@Override
		public T getResult() throws InterruptedException, ExecutionException
		{
			return get();
		}

		/**
		 * Waits if necessary for at most the given time for the computation to complete, and then retrieves its result, if
		 * available.
		 * 
		 * @see FutureTask#get(long, TimeUnit)
		 */
		@Override
		public T getResult(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
		{
			return get(timeout, unit);
		}

		/**
		 * Waits if necessary for the computation to complete, and then retrieves API call info.
		 * 
		 * @see FutureTask#get()
		 */
		@Override
		public List<ApiInvokeInfo> getInvokeInfo()
		{
			if (isDone() || isCancelled()) {
				return apiInvokeInfo;
			}
			try {
				awaitDoneHack.invoke(this, false, 0L);
			}
			catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
				throw new BugError(awaitDoneHack.toString() + " cannot be accessed?", e);
			}
			return apiInvokeInfo;
		}

		/**
		 * Waits if necessary for at most the given time for the computation to complete, and then retrieves API call info, if
		 * available.
		 * 
		 * @see FutureTask#get(long, TimeUnit)
		 */
		@Override
		public List<ApiInvokeInfo> getInvokeInfo(long timeout, TimeUnit unit)
		{
			if (isDone() || isCancelled()) {
				return apiInvokeInfo;
			}

			try {
				awaitDoneHack.invoke(this, true, unit.toNanos(timeout));
			}
			catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
				throw new BugError(awaitDoneHack.toString() + " cannot be accessed?", e);
			}

			return apiInvokeInfo;
		}

		@Override
		protected void done()
		{
			super.done();
			apiInvokeInfo = TBApiContextCache.get().getAndCleanApiInvokeInfo();
		}

		public static List<ApiInvokeInfo> getAllMergedApiInfo(ApiFuture<?>... apiFutures)
		{
			List<ApiInvokeInfo> result = new ArrayList<>();
			for (ApiFuture<?> f : apiFutures) {
				if (f.getInvokeInfo() != null) {
					result.addAll(f.getInvokeInfo());
				}
			}
			return result;
		}
	}

	private static class ApiThreadPoolExecutor extends ThreadPoolExecutor
	{

		/**
		 * @see ThreadPoolExecutor#ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue, ThreadFactory,
		 *      RejectedExecutionHandler)
		 * @param corePoolSize
		 * @param maximumPoolSize
		 * @param keepAliveTime
		 * @param unit
		 * @param workQueue
		 * @param threadFactory
		 * @param handler
		 */
		public ApiThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
				BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
		{
			super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
		}

		public <T> ApiFuture<T> subumitApiCallable(Callable<T> task)
		{
			if (task == null)
				throw new NullPointerException();
			ApiFutureTask<T> ftask = new ApiFutureTask<T>(task);
			execute(ftask);
			return ftask;
		}
	}

	private static final AtomicLong THREAD_ID = new AtomicLong();

	private static final AtomicInteger SERVICE_ID = new AtomicInteger();

	private static final ThreadFactory THREAD_FACTORY = new ThreadFactory() {
		@Override
		public Thread newThread(Runnable r)
		{
			Thread t = new Thread(r);
			t.setName("APIExecutorService-" + SERVICE_ID.incrementAndGet() + "-Thread#"
					+ ApiExecutorService.THREAD_ID.getAndIncrement());
			t.setDaemon(true);
			return t;
		}
	};

	private final ApiThreadPoolExecutor executor;

	private final long keepAliveTimeInMilliseconds = 30 * 60 * 1000;

	/**
	 * Default
	 * 
	 * @param corePoolSize
	 * @param maximumPoolSize
	 */
	public ApiExecutorService(int corePoolSize, int maximumPoolSize)
	{
		this(corePoolSize, maximumPoolSize, ApiExecutorPolicy.CALLER_RUN_POLICY);
	}

	/**
	 * @param corePoolSize
	 *            {@link ThreadPoolExecutor#getCorePoolSize()}
	 * @param maximumPoolSize
	 *            {@link ThreadPoolExecutor#getMaximumPoolSize()}
	 * @param policy
	 *            see {@link ApiExecutorPolicy}
	 */
	public ApiExecutorService(int corePoolSize, int maximumPoolSize, ApiExecutorPolicy policy)
	{
		RejectedExecutionHandler handler = null;

		if (policy == ApiExecutorPolicy.CALLER_RUN_POLICY) {
			handler = new ThreadPoolExecutor.CallerRunsPolicy();
		}
		else if (policy == ApiExecutorPolicy.REJECT_EXCEPTION_POLICY) {
			handler = new ThreadPoolExecutor.AbortPolicy();
		}

		handler = handler == null ? new ThreadPoolExecutor.CallerRunsPolicy() : handler;
		executor = new ApiThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTimeInMilliseconds, TimeUnit.MILLISECONDS,
				new SynchronousQueue<Runnable>(), THREAD_FACTORY, handler);
	}

	/**
	 * @see java.util.concurrent.AbstractExecutorService#submit(Callable)
	 * @param callable
	 * @return
	 */
	public <T> ApiFuture<T> subumit(Callable<T> callable)
	{
		return executor.subumitApiCallable(callable);
	}

	public void shutdown()
	{
		executor.shutdown();
	}
}
